package cn.touna.jss.modules.custom.carmodel.synctask;

import cn.touna.risk.data.hadoop.common.properties.HbaseProperties;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;

/**
 * Created by Steve.Tao on 2017/11/17.
 */
@Configuration
public class ZkConfiguration implements Watcher {

    private ZooKeeper zooKeeper;

    @Autowired
    private HbaseProperties hbaseProperties;

    private final static Logger logger = LogManager.getLogger();
    private CountDownLatch connectedSignal;


    @PostConstruct
    public void init() {
        String[] hosts = hbaseProperties.getZkhost().split(",");
        StringBuilder sb = new StringBuilder();
        for (String host : hosts) {
            sb.append(host).append(":").append(hbaseProperties.getZkport()).append(",");
        }
        if (sb.charAt(sb.length() - 1) == ',')
            sb.deleteCharAt(sb.length() - 1);
        String connectionString = sb.toString();
        try {
            connectedSignal = new CountDownLatch(1);
            zooKeeper = new ZooKeeper(connectionString, 60000, this);
            connectedSignal.await();
        } catch (IOException e) {
            logger.error("创建ZooKeeper客户端失败", e);
            connectedSignal.countDown();
            if(zooKeeper != null) {
                try {
                    zooKeeper.close();
                } catch (InterruptedException e1) {
                    logger.error(e.getMessage(),e1);
                }
            }
        } catch (InterruptedException e) {
            logger.error(e.getMessage(),e);
            if(zooKeeper != null) {
                try {
                    zooKeeper.close();
                } catch (InterruptedException e1) {
                    logger.error(e.getMessage(),e1);
                }
            }
        }
    }

    @Bean
    public ZooKeeper getZooKeeper() {
        if (zooKeeper == null || zooKeeper.getState() == ZooKeeper.States.CLOSED)
            init();
        return zooKeeper;
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getState() == Event.KeeperState.SyncConnected) {
            logger.info("Zookeeper连接成功");
            connectedSignal.countDown();
        }
    }
}
